package com.rocketmq.example;

import java.util.Date;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

public class E11_FiltrationMessage {
    private static final String TOPIC = "FiltrationMessage";

    private static final String GROUP = "FiltrationMessage";

    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP);
        try {
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            // consumer.subscribe(TOPIC, "*");
            consumer.subscribe(TOPIC, MessageSelector.byTag(ExampleMQConfig.TAG.getName()));
            // consumer.subscribe(TOPIC, MessageSelector.bySql("(TAGS is not null and TAGS in ('"+ExampleMQConfig.TAG.getName()+"'))  "));
            // consumer.subscribe(TOPIC, MessageSelector.bySql("(TAGS is not null and TAGS in ('"+ExampleMQConfig.TAG.getName()+"')) and (id is not null and id between 1 and 6 ) "));
            

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerOrderly() {

                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    msgs.forEach(msg -> {
                        System.out.println(
                                "消费消息:%s" + Thread.currentThread().getName() + ":" + new String(msg.getBody())
                                        + new Date());
                    });
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
            System.err.println("消费者启动成功");
        } catch (Exception e) {
            e.printStackTrace();
        } finally{
            // consumer.shutdown();
        }
    }
}
